package com.example.klinemqflink.model;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * Custom SourceFunction to emit market data
 *
 */
public class MarketDataSource implements SourceFunction<MarketData> {
    private  MarketData marketData;
    private volatile boolean isRunning = true;

    public MarketDataSource(MarketData marketData) {
        this.setMarketData(marketData);
    }

    @Override
    public void run(SourceContext<MarketData> ctx) throws Exception {
        while (isRunning) {
            ctx.collect(marketData);
            isRunning = false; // Emit only once
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    public void setMarketData(MarketData marketData) {
        this.marketData = marketData;
    }
}
